package com.lty.bbapp.kafka;

import java.util.Properties;


import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
/***
 * 
* <p>Title: KafkaProducer</p>
* <p>Description:kafka生产者 </p>
* <p>Company: lty</p>
* @author liuhao
* @date 2016年12月13日 下午6:34:39
* version 1.0
 */
public class KafkaProducer{
    private final kafka.javaapi.producer.Producer<String, String> producer;
    private final String topic;
    private final Properties props = new Properties();
    
    public static int consumeMsgCount = 0;

    /*public KafkaProducer(String topic) {
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        props.put("metadata.broker.list", "127.0.0.1:9092"); // 配置kafka端口
        
        props.put("request.required.acks", "-1");
        producer = new kafka.javaapi.producer.Producer<String, String>(new ProducerConfig(props));
        this.topic = topic;
    }*/
    
    public KafkaProducer(String topic, String kafkaServerAndPort) {
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        props.put("metadata.broker.list", kafkaServerAndPort); // 配置kafka端口
        
        props.put("request.required.acks", "-1");
        producer = new kafka.javaapi.producer.Producer<String, String>(new ProducerConfig(props));
        this.topic = topic;
    }

    //推送消息
    public void pushMsg(String busInfo) {
        try {
			producer.send(new KeyedMessage<String, String>(topic,busInfo));
		} catch (Exception e) {
			e.printStackTrace();
		}
    }
    
}

